package mysql;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @Description mysql sink
 * @Author jiangxiaozhi
 * @Date 2018/10/15 18:31
 **/
public class JdbcWriter extends RichSinkFunction<Tuple2<String,String>> {
    private Connection connection;
    private PreparedStatement preparedStatement;
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 加载JDBC驱动
        Class.forName(ConfigKeys.DRIVER_CLASS);
        // 获取数据库连接
        connection = DriverManager.getConnection(ConfigKeys.SOURCE_DRIVER_URL,ConfigKeys.SOURCE_USER,ConfigKeys.SOURCE_PASSWORD);//写入mysql数据库
        preparedStatement = connection.prepareStatement("INSERT INTO goods_info_sink(goods_id,goods_name) values(?,?)");//insert sql在配置文件中
        super.open(parameters);
    }

    @Override
    public void close() throws Exception {
        super.close();
        if(preparedStatement != null){
            preparedStatement.close();
        }
        if(connection != null){
            connection.close();
        }
        super.close();
    }

    @Override
    public void invoke(Tuple2<String,String> value, Context context) throws Exception {
        try {
            String id = value.f0;//获取JdbcReader发送过来的结果
            String name = value.f1;
            preparedStatement.setString(1,id);
            preparedStatement.setString(2,name);
            preparedStatement.executeUpdate();
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}
